package org.cancer.service

import org.apache.spark.sql.{DataFrame, SaveMode}
import org.cancer.util.SparkUtil
import org.apache.spark.sql.functions._

class SmokeCancersService$Laurel {
  def dataAnalysis(data: DataFrame): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._ //隐式转换

    // 数据预处理
    val filteredDF: DataFrame = data
      .filter($"吸烟状态".isin("从不", "曾经", "偶尔", "当前")) // 筛选有效吸烟状态
      .filter($"肿瘤类型".isNotNull) // 确保肿瘤类型不为空

    // 创建组合键并统计数量
    val resultDF = filteredDF
      .withColumn("组合键", concat($"吸烟状态", lit("_"), $"肿瘤类型"))
      .withColumn("值", lit(1))
      .groupBy("组合键", "吸烟状态", "肿瘤类型")
      .agg(count("值").as("患者数量"))
      .orderBy($"吸烟状态", desc("患者数量"))
      .select("吸烟状态","肿瘤类型","患者数量")

    // 存入数据库
    resultDF.write
      .format("jdbc") // 指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") // 链接本地数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") // mysql 8.0以上驱动
      .option("user", "root") // 用户名
      .option("password", "123456") // 密码
      .option("dbtable", "smoking_cancers_laurel") // 表名
      .mode(SaveMode.Append) // 当数据不存在的时候会自动创建
      .save()

  }
}
